package cn._51doit.kafka.clients

import java.util.{Properties, UUID}

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
import org.apache.kafka.common.serialization.StringSerializer

/**
 * 演示Kafka客户端的事务
 *
 * 1.使用Kafka的事务，每一次Producer都有设置一个唯一的事务ID
 * 2.如果在提交事务之前，调用flush，客户端的数据会被刷新到Broker中，但是数据的状态（隔离级别）为uncommitted
 * 3.如果出现异常，客户端放弃事务，缓存到客户端的数据会被丢弃
 * 4.如果没有出现异常，客户端提交事务，客户端将缓存的数据flush到borker，并将broker中对应事务ID的数据更新状态为committed，
 * 5.如果想要实现数据的一致性，必须要求Consumer设置隔离级别，只读read_committed的数据
 */
object ProducerTransactionDemo {

  def main(args: Array[String]): Unit = {

    // 1 配置参数
    val props = new Properties()
    // 连接kafka节点
    props.setProperty("bootstrap.servers", "node-1.51doit.cn:9092,node-2.51doit.cn:9092,node-3.51doit.cn:9092")
    //指定key序列化方式
    props.setProperty("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    //指定value序列化方式
    props.setProperty("value.serializer", classOf[StringSerializer].getName) // 两种写法都行

    //如果开启事务，要设置事务ID(如果写入的数据失败了，对应事务ID的数据会被标签为未提交成功)
    props.setProperty("transactional.id", UUID.randomUUID().toString)

    //创建一个Kafka的Producer的客户端
    val producer = new KafkaProducer[String, String](props)

    //开启事务
    try {

      //初始化事务
      producer.initTransactions()
      //开始事务
      producer.beginTransaction()
      //准备数据将数据发送
      val record1 = new ProducerRecord[String, String]("test", "doit123456")
      //发送数据
      producer.send(record1)
      //在没有出现异常之前，将客户端的数据flush到broker中
      producer.flush()
      val record2 = new ProducerRecord[String, String]("test", "doit987654")
      //发送数据
      producer.send(record2)
      //抛出异常
      //val i = 1 / 0
      //准备数据将数据发送
      val record3 = new ProducerRecord[String, String]("test", "doit888888")
      //发送数据
      producer.send(record3)
      //提交事务
      producer.commitTransaction()
    } catch {
      case e: Exception => {
        e.printStackTrace()
        producer.abortTransaction() //放弃当前事务（类似数据库的回滚）
      }
    }

    //close
    producer.close()






  }
}
